package com.atguigu.flink.datastreamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Created by Smexy on 2022/12/14
 *
 *  希望把每种传感器统计的vc和写入mysql
 */
public class Demo4_Custom
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId)
            .sum("vc");

        //使用自定义的sink
        ds.print();
        ds.addSink(new MySink());


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
    }

    private static class MySink extends RichSinkFunction<WaterSensor>
    {
        String sql = "replace into ws values(?,?,?)";
        private PreparedStatement ps;

        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {

                    ps.setString(1,value.getId());
                    ps.setLong(2,value.getTs());
                    ps.setInt(3,value.getVc());

                    //执行写出
                    ps.executeUpdate();
        }

        private Connection connection;

        //以Task为单位获取连接
        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(
                "jdbc:mysql://hadoop104:3306/220730?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true",
                "root",
                "000000"
            );

            ps = connection.prepareStatement(sql);
        }

        @Override
        public void close() throws Exception {

            if (connection != null){
                connection.close();
            }

        }
    }
}
